package com.asap.demo.rete;


import com.asap.demo.ContextInfo;
import com.asap.demo.function.dealMapFunction;
import com.asap.demo.function.dealStreamProcessFunction;
import com.asap.demo.model.BeanField;
import com.asap.demo.sourcefunc.MysqlSourceFunction1;
import com.asap.demo.table.RuleParse;
import com.asap.demo.utils.Constants;
import com.asap.demo.utils.Utils;
import com.asap.rule.StandardEvent;
import com.asap.rule.orm.DbFetcher;
import com.asap.rule.util.PropTransformMap;
import com.asap.rule.util.RuleReader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class ReteDemo5 {

	private static final Logger logger = LoggerFactory.getLogger(ReteDemo5.class);

	//kafka-run-class kafka.tools.ConsumerOffsetChecker --zookeeper 192.168.1.25:1813 --group temporal --topic flink_pressure_test6

	public static void main(String[] args) throws Exception {

		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
				.useBlinkPlanner()
				.inStreamingMode()
				.build();
		StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);


		env.setParallelism(1);
		env.enableCheckpointing(5000);  //检查点 每5000ms
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


		blinkStreamTableEnv.executeSql(Utils.read("/home/asap/wbh/conf/createTable.cfg"));
		Table table = blinkStreamTableEnv.from("asap_superset");

		table.printSchema();

	}

}
